/*
 * brinfuncs.c
 *		调查 BRIN 索引的函数
 *
 * Copyright (c) 2014-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		contrib/pageinspect/brinfuncs.c
 */
#include "postgres.h"

#include "access/brin.h"
#include "access/brin_internal.h"
#include "access/brin_page.h"
#include "access/brin_revmap.h"
#include "access/brin_tuple.h"
#include "access/htup_details.h"
#include "catalog/index.h"
#include "catalog/pg_am_d.h"
#include "catalog/pg_type.h"
#include "funcapi.h"
#include "lib/stringinfo.h"
#include "miscadmin.h"
#include "pageinspect.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"

PG_FUNCTION_INFO_V1(brin_page_type);
PG_FUNCTION_INFO_V1(brin_page_items);
PG_FUNCTION_INFO_V1(brin_metapage_info);
PG_FUNCTION_INFO_V1(brin_revmap_data);

#define IS_BRIN(r) ((r)->rd_rel->relam == BRIN_AM_OID)

typedef struct brin_column_state
{
	int			nstored;
	FmgrInfo	outputFn[FLEXIBLE_ARRAY_MEMBER];
} brin_column_state;


static Page fc_verify_brin_page(bytea *fc_raw_page, uint16 fc_type,
							 const char *fc_strtype);

Datum brin_page_type(PG_FUNCTION_ARGS)
{
	bytea	   *fc_raw_page = PG_GETARG_BYTEA_P(0);
	Page		fc_page;
	char	   *fc_type;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	fc_page = get_page_from_raw(fc_raw_page);

	if (PageIsNew(fc_page))
		PG_RETURN_NULL();

	/* 验证特殊空间的大小是否符合预期 */
	if (PageGetSpecialSize(fc_page) != MAXALIGN(sizeof(BrinSpecialSpace)))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("input page is not a valid %s page", "BRIN"),
				 errdetail("Expected special size %d, got %d.",
						   (int) MAXALIGN(sizeof(BrinSpecialSpace)),
						   (int) PageGetSpecialSize(fc_page))));

	switch (BrinPageType(fc_page))
	{
		case BRIN_PAGETYPE_META:
			fc_type = "meta";
			break;
		case BRIN_PAGETYPE_REVMAP:
			fc_type = "revmap";
			break;
		case BRIN_PAGETYPE_REGULAR:
			fc_type = "regular";
			break;
		default:
			fc_type = psprintf("unknown (%02x)", BrinPageType(fc_page));
			break;
	}

	PG_RETURN_TEXT_P(cstring_to_text(fc_type));
}

/*
 * 验证给定的 bytea 是否包含指定页面类型的 BRIN 页面，
 * 如果试图失败则终止。返回指向页面的指针。
 */
static Page fc_verify_brin_page(bytea *fc_raw_page, uint16 fc_type, const char *fc_strtype)
{
	Page		fc_page = get_page_from_raw(fc_raw_page);

	if (PageIsNew(fc_page))
		return fc_page;

	/* 验证特殊空间的大小是否符合预期 */
	if (PageGetSpecialSize(fc_page) != MAXALIGN(sizeof(BrinSpecialSpace)))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("input page is not a valid %s page", "BRIN"),
				 errdetail("Expected special size %d, got %d.",
						   (int) MAXALIGN(sizeof(BrinSpecialSpace)),
						   (int) PageGetSpecialSize(fc_page))));

	/* 验证特殊空间是否表示此页面是我们想要的 */
	if (BrinPageType(fc_page) != fc_type)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("page is not a BRIN page of type \"%s\"", fc_strtype),
				 errdetail("Expected special type %08x, got %08x.",
						   fc_type, BrinPageType(fc_page))));

	return fc_page;
}


/*
 * 从 BRIN 索引页面中提取所有项值
 *
 * 用法：SELECT * FROM brin_page_items(get_raw_page('idx', 1), 'idx'::regclass);
 */
Datum brin_page_items(PG_FUNCTION_ARGS)
{
	bytea	   *fc_raw_page = PG_GETARG_BYTEA_P(0);
	Oid			fc_indexRelid = PG_GETARG_OID(1);
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	Relation	fc_indexRel;
	brin_column_state **fc_columns;
	BrinDesc   *fc_bdesc;
	BrinMemTuple *fc_dtup;
	Page		fc_page;
	OffsetNumber fc_offset;
	AttrNumber	fc_attno;
	bool		fc_unusedItem;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	InitMaterializedSRF(fcinfo, 0);

	fc_indexRel = index_open(fc_indexRelid, AccessShareLock);

	if (!IS_BRIN(fc_indexRel))
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("\"%s\" is not a %s index",
						RelationGetRelationName(fc_indexRel), "BRIN")));

	fc_bdesc = brin_build_desc(fc_indexRel);

	/* 最小验证我们获得的页面 */
	fc_page = fc_verify_brin_page(fc_raw_page, BRIN_PAGETYPE_REGULAR, "regular");

	if (PageIsNew(fc_page))
	{
		brin_free_desc(fc_bdesc);
		index_close(fc_indexRel, AccessShareLock);
		PG_RETURN_NULL();
	}

	/*
	 * 为所有索引数据类型初始化输出函数；简化
	 * 后续的调用。
	 */
	fc_columns = palloc(sizeof(brin_column_state *) * RelationGetDescr(fc_indexRel)->natts);
	for (fc_attno = 1; fc_attno <= fc_bdesc->bd_tupdesc->natts; fc_attno++)
	{
		Oid			fc_output;
		bool		fc_isVarlena;
		BrinOpcInfo *fc_opcinfo;
		int			fc_i;
		brin_column_state *fc_column;

		fc_opcinfo = fc_bdesc->bd_info[fc_attno - 1];
		fc_column = palloc(offsetof(brin_column_state, outputFn) +
						sizeof(FmgrInfo) * fc_opcinfo->oi_nstored);

		fc_column->nstored = fc_opcinfo->oi_nstored;
		for (fc_i = 0; fc_i < fc_opcinfo->oi_nstored; fc_i++)
		{
			getTypeOutputInfo(fc_opcinfo->oi_typcache[fc_i]->type_id, &fc_output, &fc_isVarlena);
			fmgr_info(fc_output, &fc_column->outputFn[fc_i]);
		}

		fc_columns[fc_attno - 1] = fc_column;
	}

	fc_offset = FirstOffsetNumber;
	fc_unusedItem = false;
	fc_dtup = NULL;
	for (;;)
	{
		Datum		fc_values[7];
		bool		fc_nulls[7];

		/*
		 * 这个循环对页面中每个元组的每个属性调用一次。
		 * 在元组开始时，我们获取一个 NULL dtup；这就是我们
		 * 获取和解码下一个元组的信号。如果不是这种情况，
		 * 我们输出下一个属性。
		 */
		if (fc_dtup == NULL)
		{
			ItemId		fc_itemId;

			/* 验证项目状态：如果没有数据，我们无法解码 */
			fc_itemId = PageGetItemId(fc_page, fc_offset);
			if (ItemIdIsUsed(fc_itemId))
			{
				fc_dtup = brin_deform_tuple(fc_bdesc,
										 (BrinTuple *) PageGetItem(fc_page, fc_itemId),
										 NULL);
				fc_attno = 1;
				fc_unusedItem = false;
			}
			else
				fc_unusedItem = true;
		}
		else
			fc_attno++;

		MemSet(fc_nulls, 0, sizeof(fc_nulls));

		if (fc_unusedItem)
		{
			fc_values[0] = UInt16GetDatum(fc_offset);
			fc_nulls[1] = true;
			fc_nulls[2] = true;
			fc_nulls[3] = true;
			fc_nulls[4] = true;
			fc_nulls[5] = true;
			fc_nulls[6] = true;
		}
		else
		{
			int			fc_att = fc_attno - 1;

			fc_values[0] = UInt16GetDatum(fc_offset);
			switch (TupleDescAttr(fc_rsinfo->setDesc, 1)->atttypid)
			{
				case INT8OID:
					fc_values[1] = Int64GetDatum((int64) fc_dtup->bt_blkno);
					break;
				case INT4OID:
					/* 对旧扩展版本的支持 */
					fc_values[1] = UInt32GetDatum(fc_dtup->bt_blkno);
					break;
				default:
					elog(ERROR, "incorrect output types");
			}
			fc_values[2] = UInt16GetDatum(fc_attno);
			fc_values[3] = BoolGetDatum(fc_dtup->bt_columns[fc_att].bv_allnulls);
			fc_values[4] = BoolGetDatum(fc_dtup->bt_columns[fc_att].bv_hasnulls);
			fc_values[5] = BoolGetDatum(fc_dtup->bt_placeholder);
			if (!fc_dtup->bt_columns[fc_att].bv_allnulls)
			{
				BrinValues *fc_bvalues = &fc_dtup->bt_columns[fc_att];
				StringInfoData fc_s;
				bool		fc_first;
				int			fc_i;

				initStringInfo(&fc_s);
				appendStringInfoChar(&fc_s, '{');

				fc_first = true;
				for (fc_i = 0; fc_i < fc_columns[fc_att]->nstored; fc_i++)
				{
					char	   *fc_val;

					if (!fc_first)
						appendStringInfoString(&fc_s, " .. ");
					fc_first = false;
					fc_val = OutputFunctionCall(&fc_columns[fc_att]->outputFn[fc_i],
											 fc_bvalues->bv_values[fc_i]);
					appendStringInfoString(&fc_s, fc_val);
					pfree(fc_val);
				}
				appendStringInfoChar(&fc_s, '}');

				fc_values[6] = CStringGetTextDatum(fc_s.data);
				pfree(fc_s.data);
			}
			else
			{
				fc_nulls[6] = true;
			}
		}

		tuplestore_putvalues(fc_rsinfo->setResult, fc_rsinfo->setDesc, fc_values, fc_nulls);

		/*
		 * 如果该项未被使用，直接跳到下一个；否则，
		 * 唯一需要的清理是通过释放当前项将我们的信号
		 * 设置为在下一次迭代中转到下一个元组。
		 */
		if (fc_unusedItem)
			fc_offset = OffsetNumberNext(fc_offset);
		else if (fc_attno >= fc_bdesc->bd_tupdesc->natts)
		{
			pfree(fc_dtup);
			fc_dtup = NULL;
			fc_offset = OffsetNumberNext(fc_offset);
		}

		/*
		 * 如果我们超出了页面末尾，我们就完成了。
		 */
		if (fc_offset > PageGetMaxOffsetNumber(fc_page))
			break;
	}

	brin_free_desc(fc_bdesc);
	index_close(fc_indexRel, AccessShareLock);

	return (Datum) 0;
}

Datum brin_metapage_info(PG_FUNCTION_ARGS)
{
	bytea	   *fc_raw_page = PG_GETARG_BYTEA_P(0);
	Page		fc_page;
	BrinMetaPageData *fc_meta;
	TupleDesc	fc_tupdesc;
	Datum		fc_values[4];
	bool		fc_nulls[4];
	HeapTuple	fc_htup;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	fc_page = fc_verify_brin_page(fc_raw_page, BRIN_PAGETYPE_META, "metapage");

	if (PageIsNew(fc_page))
		PG_RETURN_NULL();

	/* 为我们的结果类型构建一个元组描述符 */
	if (get_call_result_type(fcinfo, NULL, &fc_tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");
	fc_tupdesc = BlessTupleDesc(fc_tupdesc);

	/* 从元数据页中提取值 */
	fc_meta = (BrinMetaPageData *) PageGetContents(fc_page);
	MemSet(fc_nulls, 0, sizeof(fc_nulls));
	fc_values[0] = CStringGetTextDatum(psprintf("0x%08X", fc_meta->brinMagic));
	fc_values[1] = Int32GetDatum(fc_meta->brinVersion);
	fc_values[2] = Int32GetDatum(fc_meta->pagesPerRange);
	fc_values[3] = Int64GetDatum(fc_meta->lastRevmapPage);

	fc_htup = heap_form_tuple(fc_tupdesc, fc_values, fc_nulls);

	PG_RETURN_DATUM(HeapTupleGetDatum(fc_htup));
}

/*
 * 返回存储在 BRIN 反向映射页面中的 TID 数组
 */
Datum brin_revmap_data(PG_FUNCTION_ARGS)
{
	struct
	{
		ItemPointerData *tids;
		int			idx;
	}		   *fc_state;
	FuncCallContext *fc_fctx;

	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be superuser to use raw page functions")));

	if (SRF_IS_FIRSTCALL())
	{
		bytea	   *fc_raw_page = PG_GETARG_BYTEA_P(0);
		MemoryContext fc_mctx;
		Page		fc_page;

		/* 为跨调用持久性创建一个函数上下文 */
		fc_fctx = SRF_FIRSTCALL_INIT();

		/* 切换到适合多个函数调用的内存上下文 */
		fc_mctx = MemoryContextSwitchTo(fc_fctx->multi_call_memory_ctx);

		/* 最小验证我们获得的页面 */
		fc_page = fc_verify_brin_page(fc_raw_page, BRIN_PAGETYPE_REVMAP, "revmap");

		if (PageIsNew(fc_page))
		{
			MemoryContextSwitchTo(fc_mctx);
			PG_RETURN_NULL();
		}

		fc_state = palloc(sizeof(*fc_state));
		fc_state->tids = ((RevmapContents *) PageGetContents(fc_page))->rm_tids;
		fc_state->idx = 0;

		fc_fctx->user_fctx = fc_state;

		MemoryContextSwitchTo(fc_mctx);
	}

	fc_fctx = SRF_PERCALL_SETUP();
	fc_state = fc_fctx->user_fctx;

	if (fc_state->idx < REVMAP_PAGE_MAXITEMS)
		SRF_RETURN_NEXT(fc_fctx, PointerGetDatum(&fc_state->tids[fc_state->idx++]));

	SRF_RETURN_DONE(fc_fctx);
}
